package com.five.consumer;

import com.alibaba.fastjson.JSON;
import com.five.mapper.CollectionMapper;
import com.five.message.CollectMessage;
import com.five.pojo.Collection;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.nio.charset.StandardCharsets;
import java.util.Date;


@Service
@Slf4j
@RocketMQMessageListener(
        topic = "reader_topic",
        selectorExpression = "collect_add",
        consumerGroup = "reader-service-collect-consumer-group"
)
public class AddCollectConsumer implements RocketMQListener<MessageExt> { // 改为接收MessageExt
    @Autowired
    private CollectionMapper collectMapper;

    @Override
    @Transactional
    public void onMessage(MessageExt messageExt) {
        try {
            // 1. 解析消息体
            byte[] body = messageExt.getBody();
            CollectMessage msg = JSON.parseObject(body, CollectMessage.class);
            if (msg == null) {
                log.error("新增收藏消息体解析失败，payload: {}", new String(body, StandardCharsets.UTF_8));
                return;
            }

            // 2. 只处理新增操作（operateType=1）
            if (msg.getOperateType() != 1) {
                log.debug("非新增收藏消息，忽略处理：operateType={}", msg.getOperateType());
                return;
            }

            // 3. 获取关键参数（优先消息体，其次消息头）
            Long readerId = msg.getReaderId();
            Long novelId = msg.getNovelId();

            // 从消息头补充参数（若消息体未包含）
            if (readerId == null) {
                String readerIdStr = messageExt.getUserProperty("readerId");
                if (readerIdStr != null) {
                    readerId = Long.valueOf(readerIdStr);
                }
            }
            if (novelId == null) {
                String novelIdStr = messageExt.getUserProperty("novelId");
                if (novelIdStr != null) {
                    novelId = Long.valueOf(novelIdStr);
                }
            }

            // 校验参数完整性
            if (readerId == null || novelId == null) {
                log.warn("新增收藏消息缺少关键参数：readerId={}, novelId={}，忽略消息", readerId, novelId);
                return;
            }

            log.info("开始处理新增收藏消息：readerId={}, novelId={}", readerId, novelId);

            // 4. 幂等处理：检查MySQL是否已存在记录
            int count = collectMapper.countByUserIdAndNovelId(readerId, novelId);
            if (count > 0) {
                log.warn("MySQL已存在收藏记录（幂等处理）：readerId={}, novelId={}", readerId, novelId);
                return;
            }

            // 5. 写入MySQL
            Collection collect = new Collection();
            collect.setReaderId(readerId);
            collect.setNovelId(novelId);
            collect.setCreateTime(new Date());
            collectMapper.insert(collect);
            log.info("消费新增收藏消息成功：readerId={}, novelId={}", readerId, novelId);

        } catch (Exception e) {
            log.error("消费新增收藏消息失败，将触发重试", e);
            throw e; // 抛出异常触发RocketMQ重试机制
        }
    }
}